package com.enjoyor.soa.traffic.util.handler;

import java.lang.reflect.Constructor;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.TimeZone;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

import org.apache.log4j.Logger;

/**
 * 带缓冲池的消息处理器类。<br>
 * 此处理类基于线程，一直监听着缓冲池，一旦有消息进入缓冲池，立即唤醒处理器处理消息
<hr><pre>例1：
BasicMsgHandler h = new BasicMsgHandler&lt;Xxx&gt;(){
    &#64;Override
    protected boolean process(Xxx msg) {
        // 重载此函数（必须）。消息处理...
        return true;
    }
};
//将消息加入缓冲池
Xxx m=new Xxx();
h.handle(m);
</pre><hr>
<pre>例2，消息处理链(A-&gt;BC |  B-&gt;D)：
BasicMsgHandler handlerA = new BasicMsgHandler&lt;Xxx&gt;("处理器A"){ &#64;Override protected boolean process(Xxx msg) { &#47;*...*&#47; return true; } };
BasicMsgHandler handlerB = new BasicMsgHandler&lt;Xxx&gt;("处理器B"){ &#64;Override protected boolean process(Xxx msg) { &#47;*...*&#47; return true; } };
BasicMsgHandler handlerC = new BasicMsgHandler&lt;Xxx&gt;("处理器C"){ &#64;Override protected boolean process(Xxx msg) { &#47;*...*&#47; return true; } };
BasicMsgHandler handlerD = new BasicMsgHandler&lt;Xxx&gt;("处理器D"){ &#64;Override protected boolean process(Xxx msg) { &#47;*...*&#47; return true; } };
handlerA.addNextHandler(handlerB);
handlerA.addNextHandler(handlerC);
handlerB.addNextHandler(handlerD);
//将消息加入缓冲池，handlerA处理完成后，自动执行后续流程
Xxx m=new Xxx();
handlerA.handle(m);
</pre><hr>
<pre>例3，多线程处理器、单例模式：
public class MyMultiProcessHandler extends BasicMsgHandler&lt;Xxx&gt;{
    //单例模式，私有化构造函数防止该类在外部被实例化；获取单例:MyMultiProcessHandler.getInstance(MyMultiProcessHandler.class)
    private MyMultiProcessHandler(){
        super("MyMultiProcessHandler处理器");
        this.setMultiProcess(3, 5, 20);//多线程处理
    }
    &#64;Override protected boolean process(Xxx msg) {
        // 重载此函数（必须）。消息处理...
        return true;
    }
}
//将消息加入缓冲池
Xxx m=new Xxx();
MyMultiProcessHandler.getInstance(MyMultiProcessHandler.class).handle(m);
</pre><hr>
<pre>例4，与监听类结合（将监听的消息放入处理队列不阻塞监听程序）：
监听类接口：
public interface XxxListener{
    public void notify(Xxx x);
}
与监听类结合的消息处理
public class MyXxxListenerHandler extends BasicMsgHandler&lt;Xxx&gt; implements XxxListener{
    &#64;Override 
    protected boolean process(Xxx msg) {
        // 重载此函数（必须）。消息处理...
        return true;
    }
    &#64;Override 
    public void notify(Xxx msg) {
        this.handle(msg); //将消息加入缓冲池
    }
}
</pre><hr>
 * @param <M> 消息类型
 */
@SuppressWarnings({"rawtypes","unchecked"})
abstract public class BasicMsgHandler<M> extends Thread{
	private static Logger log = Logger.getLogger(BasicMsgHandler.class);
	
	
    /**
     * 获取单例
     * <hr>
<pre>例：
public class MyInstanceHandler extends BasicMsgHandler&lt;Xxx&gt;{
    //私有化构造函数防止该类在外部被实例化
    private MyInstanceHandler(){ }
    &#64;Override 
    protected boolean process(Xxx msg) { &#47;*重载此函数（必须）。消息处理...*&#47; return true; }
};

获取单例:MyInstanceHandler.getInstance(MyInstanceHandler.class)

</pre><hr>
     * @param type
     * @return
     */
	public static<T extends BasicMsgHandler> T getInstance(Class<T> type){
    	if(!mapInstance.containsKey(type)){synchronized (mapInstance){if(!mapInstance.containsKey(type)){
    		try{
	    		Constructor<T> constructor = type.getDeclaredConstructor(new Class[]{});
	    		constructor.setAccessible(true);
	    		T instance = constructor.newInstance();
	    		mapInstance.put(type, instance);
    		}catch(Throwable e){
    			e.printStackTrace();
    		}
    	}}}
    	return (T) mapInstance.get(type);
    }
	protected static Map<Class<?>,Object> mapInstance=new HashMap<Class<?>,Object>();

	
	/**
	 * @param cacheLength  队列长度。默认：500。
	 * @param handlerName  处理器名称。默认："数据处理器"。
	 * @param blockDelete 添加时是否不阻塞。默认：false。
	 * @param priority  是否优先级队列。默认：false。如果是优先级，则 M msg 必须实现Comparable接口！
	 */
	protected BasicMsgHandler(){
		 this(500,null,false);
	}
	/**
	 * @param cacheLength  队列长度。默认：500。
	 * @param handlerName  处理器名称。默认："数据处理器"。
	 * @param blockDelete 添加时是否不阻塞。默认：false。
	 * @param priority  是否优先级队列。默认：false。如果是优先级，则 M msg 必须实现Comparable接口！
	 */
	protected BasicMsgHandler(boolean blockDelete){
		 this(500,null,blockDelete);
	}
	/**
	 * @param cacheLength  队列长度。默认：500。
	 * @param handlerName  处理器名称。默认："数据处理器"。
	 * @param blockDelete 添加时是否不阻塞。默认：false。
	 * @param priority  是否优先级队列。默认：false。如果是优先级，则 M msg 必须实现Comparable接口！
	 */
	protected BasicMsgHandler(int cacheLength){
		this(cacheLength,null,false);
	}
	/**
	 * @param cacheLength  队列长度。默认：500。
	 * @param handlerName  处理器名称。默认："数据处理器"。
	 * @param blockDelete 添加时是否不阻塞。默认：false。
	 * @param priority  是否优先级队列。默认：false。如果是优先级，则 M msg 必须实现Comparable接口！
	 */
	protected BasicMsgHandler(String handlerName){
		this(500,handlerName,false);
	}
	/**
	 * @param cacheLength  队列长度。默认：500。
	 * @param handlerName  处理器名称。默认："数据处理器"。
	 * @param blockDelete 添加时是否不阻塞。默认：false。
	 * @param priority  是否优先级队列。默认：false。如果是优先级，则 M msg 必须实现Comparable接口！
	 */
	protected BasicMsgHandler(int cacheLength,boolean blockDelete){
		this(cacheLength,null,blockDelete);
	}
	/**
	 * @param cacheLength  队列长度。默认：500。
	 * @param handlerName  处理器名称。默认："数据处理器"。
	 * @param blockDelete 添加时是否不阻塞。默认：false。
	 * @param priority  是否优先级队列。默认：false。如果是优先级，则 M msg 必须实现Comparable接口！
	 */
	protected BasicMsgHandler(String handlerName,boolean blockDelete){
		this(500,handlerName,blockDelete);
	}
	/**
	 * @param cacheLength  队列长度。默认：500。
	 * @param handlerName  处理器名称。默认："数据处理器"。
	 * @param blockDelete 添加时是否不阻塞。默认：false。
	 * @param priority  是否优先级队列。默认：false。如果是优先级，则 M msg 必须实现Comparable接口！
	 */
	protected BasicMsgHandler(int cacheLength,String handlerName){
		this(cacheLength,handlerName,false);
	}
	/**
	 * @param cacheLength  队列长度。默认：500。
	 * @param handlerName  处理器名称。默认："数据处理器"。
	 * @param blockDelete 添加时是否不阻塞。默认：false。
	 * @param priority  是否优先级队列。默认：false。如果是优先级，则 M msg 必须实现Comparable接口！
	 */
	protected BasicMsgHandler(int cacheLength,String handlerName,boolean blockDelete){
		this(cacheLength,handlerName,blockDelete,false);
	}
	/**
	 * @param cacheLength  队列长度。默认：500。
	 * @param handlerName  处理器名称。默认："数据处理器"。
	 * @param blockDelete 添加时是否不阻塞。默认：false。
	 * @param priority  是否优先级队列。默认：false。如果是优先级，则消息 M msg 必须实现Comparable接口！
	 */
	protected BasicMsgHandler(Integer cacheLength,String handlerName,boolean blockDelete,boolean priority){
		handlerId = _id_count.getAndIncrement();
		this.blockDelete = blockDelete;
		this.handlerName = (handlerName==null||"".equals(handlerName)?"数据处理器":handlerName);
		cacheLength=(cacheLength==null||cacheLength<=0?500:cacheLength);
		//优先级队列
		if(priority){
			queueMsg = new PriorityBlockingQueue<M>(cacheLength);
		}else{
			queueMsg = new LinkedBlockingQueue<M>(cacheLength);
		}
		
		this.start();
		allHandlers.add(this);
		if(bmhTimer==null){synchronized(BasicMsgHandler.class){if(bmhTimer==null){
			bmhTimer=new BmhTask();
			bmhTimer.start();
		}}}
	}
	/**
	 * 数据处理器 ID
	 */
	protected Long handlerId;
	private static AtomicLong _id_count= new AtomicLong(1000);
	/**
	 * 数据处理器 ID
	 */
	public Long getHandlerId() {
		return handlerId;
	}
	
	/**
	 * 数据处理器 名称
	 */
	protected String handlerName;
	/**
	 * 数据处理器 名称
	 */
	public String getHandlerName() {
		return handlerName;
	}
	/**
	 * 数据处理器 名称
	 */
	public void setHandlerName(String handlerName) {
		this.handlerName = handlerName;
	}
	
	/**
	 * 下一步处理流程/子流程列表
	 */
	protected CopyOnWriteArrayList<BasicMsgHandler> nextHandlerList = new CopyOnWriteArrayList<BasicMsgHandler>();
	protected CopyOnWriteArrayList<BasicMsgHandler> parentHandlerList = new CopyOnWriteArrayList<BasicMsgHandler>();

	/**
	 * 停止所有数据处理器.
	 * @param wait 阻塞至全部数据处理器完成操作或超时(5分钟)。默认：false。
	 */
	public static void stopAllHandlers() {
		for(Iterator<BasicMsgHandler> iterator = BasicMsgHandler.allHandlers.iterator(); iterator.hasNext();){
            BasicMsgHandler hd = iterator.next();
            try{
            	if(hd.parentHandlerList.size()==0) hd.stopHandler();
            }catch(Throwable throwable) {hd.stopHandler();}
        }
	}
	/**
	 * 停止所有数据处理器.
	 * @param wait 阻塞至全部数据处理器完成操作或超时(5分钟)。默认：false。
	 */
	public static void stopAllHandlers(boolean wait) {
		stopAllHandlers(wait?300000:0);
	}
	/**
	 * 停止所有数据处理器.阻塞至全部数据处理器完成操作或超时
	 * @param waitTime 超时时间
	 */
	public static void stopAllHandlers(long waitTime) {
		stopAllHandlers();
		if(waitTime>0){
			try{
				synchronized(BasicMsgHandler.class){
					if(!allHandlers.isEmpty())BasicMsgHandler.class.wait(waitTime);
				}
			}catch(Throwable e){}
		}
	}
	/*public List<BasicMsgHandler> getNextHandlerList() {
		return nextHandlerList;
	}
	public void setNextHandlerList(List<BasicMsgHandler> nextHandlerList) {
		if(nextHandlerList!=null){
			for(BasicMsgHandler next:nextHandlerList){
				addNextHandler(next);
			}
		}
	}*/
	/**
	 * 停止子流程
	 * @param id 子流程ID
	 * @return 本流程含有子流程ID为id的子流程，则停止此子流程，返回true；否则返回false。
	 */
	public boolean removeNextHandler(Long id) {
		try{
			Iterator<BasicMsgHandler> it = nextHandlerList.iterator();
			while(it.hasNext()){
				BasicMsgHandler handler = it.next();
				if(handler.getHandlerId()==null || handler.getHandlerId().equals(id)){
					nextHandlerList.remove(handler);
					handler.parentHandlerList.remove(this);
					handler.stopHandler();
					return true;
				}
			}
		}catch(Throwable e){e.printStackTrace();}
		return false;
	}
	/**
	 * 添加一个“下一步处理流程”。
	 * 若handle(M msg) 返回true，则自动执行所有“下一步处理流程”
	 * @param nextHandler 下一步处理流程
	 */
	public void addNextHandler(BasicMsgHandler nextHandler) {
		if(!nextHandlerList.contains(nextHandler))
			nextHandlerList.add(nextHandler);
		if(!nextHandler.parentHandlerList.contains(this))
			nextHandler.parentHandlerList.add(this);
	}
	/**
	 * 停止所有“下一步处理流程”
	 */
	public void stopNextHandlerList() {
		Iterator<BasicMsgHandler> it = nextHandlerList.iterator();
		while(it.hasNext()){
			BasicMsgHandler handler = it.next();
			handler.stopHandler();
			it.remove();
		}
	}
	
	/**
	 * 执行下一流程；若handle(M msg),返回true,自动执行handleNext(msg)
	 * @param msg
	 * @return
	 */
	protected boolean handleNext(Object msg) {
		handlerCount.incrementAndGet();
		boolean ret = true;
		Iterator<BasicMsgHandler> it = nextHandlerList.iterator();
		while(it.hasNext()){
			BasicMsgHandler handler = it.next();
			ret &= handler.handle(msg);
		}
		return ret;
	}
	
	/**
	 * 记录日志
	 */
	public void log(){
		long entr = entrCount.get();
        long hand = handlerCount.get();
        long erro = errorCount.get();
		if(entr>0|| hand>0||erro>0){
			log.info(this.handlerName+" "+this.handlerId + " " +
					log_format.format(new Date())+
					" 接收数据量:"+(entr-log_entr)+"/"+entr+
					" 处理数据量:"+(hand-log_hand)+"/"+hand+
					" 错误数据量:"+(erro-log_erro)+"/"+erro+
					" 队列中剩余数据量:"+queueMsg.size());
		}
        log_entr = entr;
        log_hand = hand;
        log_erro = erro;
	}
	protected void logDay(){
        log.info(this.handlerName+" "+this.handlerId + " " +
        		log_formatDay.format(new Date(task_today)) +
				" 接收数据量:"+entrCount.getAndSet(0)+
				" 处理数据量:"+handlerCount.getAndSet(0)+
				" 错误数据量:"+errorCount.getAndSet(0));
        log_entr = 0L;
        log_hand = 0L;
        log_erro = 0L;
    }
	protected static SimpleDateFormat log_formatDay = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    protected static SimpleDateFormat log_format = new SimpleDateFormat("HH:mm:ss");
    protected AtomicLong entrCount = new AtomicLong();
    protected AtomicLong handlerCount = new AtomicLong();
    protected AtomicLong errorCount = new AtomicLong();
    protected long log_entr = 0L;
    protected long log_hand = 0L;
    protected long log_erro = 0L;
    protected static CopyOnWriteArrayList<BasicMsgHandler> allHandlers = new CopyOnWriteArrayList<BasicMsgHandler>();
    static BmhTask bmhTimer;
    protected static long DAY_MILLIS = 24*60*60*1000;
    static long RawOffset = TimeZone.getDefault().getRawOffset();
    protected static long task_today = getZeroTime(new Date());
    static public long getZeroTime(Date d){return ((d.getTime() + RawOffset) / DAY_MILLIS) * DAY_MILLIS - RawOffset;}
    protected int taskTimes = 0;
	static class BmhTask extends Thread{
		@Override
		public void run() {
			long newMinutes=0l,oldMinutes=0l;
			while(!allHandlers.isEmpty()){
				long thisTime = System.currentTimeMillis();
				//每秒钟检查一次
				try{
					for(Iterator iterator = BasicMsgHandler.allHandlers.iterator(); iterator.hasNext();){
	                    BasicMsgHandler hd = (BasicMsgHandler)iterator.next();
	                    try{//多线程管理
		                    if(hd.multiProcess && !hd.bStop){
		                    	int ie=(int) hd.entrCount.get();
		                    	int ih=(int) hd.handlerCount.get();
		                    	int ic=hd.queueMsg.remainingCapacity();
		                    	int is=hd.queueMsg.size();
	                    		int needn = (is-is/10)/hd.maxMsgPerProcess+1;
		                    	if(ic<is/10)
		                    		needn=hd.maxProcesses;
		                    	else if(ic>is*9/10)
		                    		needn=hd.minProcesses;
		                    	else{
		                    		
		                    	}
		                    	
	                    		if(needn>hd.multiThread.size()+1 && hd.multiThread.size()+1<hd.maxProcesses){
	                    			for(int i=hd.multiThread.size()+1;i<=needn && i<=hd.maxProcesses;i++)hd.addProessThread();
	                    		}
	                    		if(needn<hd.multiThread.size()+1 && hd.multiThread.size()+1>hd.minProcesses){
	                    			for(int i=hd.multiThread.size()+1;i>=needn&&i>=hd.minProcesses;i--)hd.removeProessThread();
	                    		}
		                    }
	                    }catch(Throwable e){}
	                }
				}catch(Throwable throwable) { }
				
				//每分钟执行一次
				newMinutes=thisTime/60000;
                if(newMinutes!=oldMinutes){
                	oldMinutes=newMinutes;
					try{
						for(Iterator iterator = BasicMsgHandler.allHandlers.iterator(); iterator.hasNext();){
		                    BasicMsgHandler hd = (BasicMsgHandler)iterator.next();
	                    
		                    try{
		                        hd.log();
		                    }catch(Throwable throwable) { }
		                    try{
		                        if(hd.task(hd.taskTimes))
		                            hd.taskTimes = 0;
		                        else
		                            hd.taskTimes++;
		                    }catch(Throwable throwable1) { }
	                    }
					}catch(Throwable throwable) { }
                }
                //每天执行一次
				try{
					long today = BasicMsgHandler.getZeroTime(new Date());
	                if(today != BasicMsgHandler.task_today){
	                    for(Iterator iterator1 = BasicMsgHandler.allHandlers.iterator(); iterator1.hasNext();){
	                        BasicMsgHandler hd = (BasicMsgHandler)iterator1.next();
	                        try{
	                            hd.logDay();
	                        }catch(Throwable throwable2) { }
	                        try{
	                            hd.taskDay();
	                        }catch(Throwable throwable3) { }
	                    }
	
	                    BasicMsgHandler.task_today = today;
	                }
				}catch(Throwable throwable) { }
				try {
					long sleept = 1000-(System.currentTimeMillis()-thisTime);
					if(sleept>0)Thread.sleep(sleept);
				} catch (Throwable e) { }
			}
		}
	}
	/**
	 * 可重载此函数，定时每分钟执行。例子：
	 * <pre>
&#64;Override
protected boolean task(int taskTimes){
    if(taskTimes>=3){ //每隔3分钟执行一次
        //...
        return true;
    }
    return false;
}
	 * </pre>
	 * @param taskTimes 计数器
	 * @return 返回true则重置计数器为0，否则计数器一直累加
	 */
	protected boolean task(int taskTimes)
    {
		/* 例子：	每隔3分钟执行一次
			if(taskTimes>=3){
				//...
				return true;
			}
			return false;  */
        return true;
    }

	/**
	 * 重载此函数，定时每日凌晨执行
	 */
    protected void taskDay()
    {
    }
	
	
	/**
	 * 数据处理器：停止数据处理，并停止后续处理流程
	 */
	public void stopHandler(){
		bStop=true;
		// bug,等待当前处理结束后，才结束下一进程
		//stopMultiProcess();
		//stopNextHandlerList();
	}
	protected boolean bStop = false;

	/**
	 * 重载此函数，数据处理的初始化
	 */
	protected void init(){
		
	}
	/**
	 * 重载此函数，数据处理停止后的清理
	 */
	protected void stopped(){
		
	}
	
	/**
	 * 数据处理器：数据接入 <br />
	 * blockDelete=true 时防止阻塞，消息入队时，若队列满，则删除一旧数据
	 * @param msg 一条数据
	 * @return
	 */
	public boolean handle(M msg){
		try {
			if(bStop) return false;
			if(blockDelete){
				while(!queueMsg.offer(msg)){
					queueMsg.poll(5, TimeUnit.MILLISECONDS);
				}
			}else{
				queueMsg.put(msg);
			}
			entrCount.incrementAndGet();
			return true;
		} catch (Exception e) {
			e.printStackTrace();
		}
		return false;
	}
	//true时防止阻塞，消息入队时，若队列满，则删除一旧数据
	protected boolean blockDelete = false;
	/**
	 * 设置队列是否防阻塞，true时防止阻塞，消息入队时，若队列满，则删除一旧数据
	 */
	public void setBlockDelete(boolean blockDelete){
		this.blockDelete=blockDelete;
	}
	
	/**
	 * 数据缓存队列
	 */
	protected BlockingQueue<M> queueMsg;
	
	/**
	 * 重新设置队列
	 * @param cacheLength
	 * @param priority
	 */
	public void setBlockingQueue(int cacheLength,boolean priority){
		BlockingQueue<M> old=queueMsg;
		//优先级队列
		cacheLength=(cacheLength<=0?500:cacheLength);
		if(priority){
			queueMsg = new PriorityBlockingQueue<M>(cacheLength);
		}else{
			queueMsg = new LinkedBlockingQueue<M>(cacheLength);
		}
		try {
			M m;
			while((m = old.poll(100, TimeUnit.MILLISECONDS))!=null){
				if(bStop) break;
				if(blockDelete){
					while(!queueMsg.offer(m)){
						queueMsg.poll(5, TimeUnit.MILLISECONDS);
					}
				}else{
					queueMsg.put(m);
				}
			}
		} catch (Exception e) {}
	}

	
	/**
	 * 数据处理超时等待时间（毫秒）
	 */
	protected int WAIT_TIME = 1000;
	/**
	 * 最多一次性取MAX_COUNT条记录
	 */
	protected int MAX_COUNT = 300;
	
	/**
	 * 数据处理器：
	 *  数据处理,
	 *  返回true将数据流转至下一流程
	 * @param msg 一条数据
	 * @return
	 * @throws Exception 
	 */
	abstract protected boolean process(M msg) throws Exception;
	//读取并由处理程序(protected boolean process(M msg))处理信息；
	private void _processMsg(){
		try {
			M m = queueMsg.poll(100, TimeUnit.MILLISECONDS);
			if(m!=null){
				try {
					if(process(m)){
						handleNext(m);
					}
				} catch (Throwable e) {
					errorCount.incrementAndGet();
					log.error(this.handlerName + " "+this.handlerId + " " + " 消息处理错误:"+e.getMessage(),e);
				}
			}
		} catch (Throwable e) {
			log.error("消息处理错误:"+e.getMessage(),e);
		}
	}
	/**
	 * 一次性取多条记录。取了首条记录后，调用此方法一次性获取多条记录，以满足批量处理要求。<br/>
	 * 一次取记录数相关配置：最多等待时间:{@link #WAIT_TIME}、最多一次取记录条数:{@link #MAX_COUNT}
<pre>例子:
protected @Override boolean process(M msg){
	List<M> lst = this.getListMsg(msg);
	//一次处理多条记录lst...
	return true;
}
</pre>
	 * @param first 首条记录
	 * @return
	 */
	protected List<M> getListMsg(M first){
		List<M> list = new ArrayList<M>();
		list.add(first);
		long currentTime = System.currentTimeMillis();
		M more =null;
		try{
			while((!(bStop&&queueMsg.isEmpty()))&&(list.size()<MAX_COUNT && System.currentTimeMillis()-currentTime<WAIT_TIME)){
				more = queueMsg.poll(10, TimeUnit.MILLISECONDS);
				if(more!=null){
					handlerCount.incrementAndGet();
					list.add(more);
				}
			}
		}catch(Throwable e){}
		return list;
	}
	
	/**
	 * 设置多线程处理
	 */
	public void setMultiProcess(int minProcesses,int maxProcesses,int maxMsgPerProcess){
		this.minProcesses = (minProcesses>0?minProcesses:1);
		this.maxProcesses = (maxProcesses>0?maxProcesses:1);
		this.maxMsgPerProcess = (maxMsgPerProcess>0?maxMsgPerProcess:5);
		if(multiThread==null){
			multiThread=new ArrayList<ProcessThread>();
		}
		multiProcess = true;
		for(int i=1;i<this.minProcesses;i++) addProessThread();
	}
	//多处理器
	private boolean multiProcess=false;
	private List<ProcessThread> multiThread=null;
	private int minProcesses,maxProcesses,maxMsgPerProcess;
	AtomicLong atomMultID = new AtomicLong(2000);
	private class ProcessThread extends Thread{
		long multid=atomMultID.incrementAndGet();
		public ProcessThread(){ this.start(); }
		public boolean stopProcessThread=false;
		@Override
		public void run(){
			log.debug(handlerName+" "+handlerId+" 多处理器MultiProcess"+multid+" start.");
            while(!stopProcessThread && ( !bStop || !queueMsg.isEmpty()) ){
				_processMsg();
			}
            multiThread.remove(this);
			log.debug(handlerName+" "+handlerId+" 多处理器MultiProcess"+multid+" stop.");
		}
	}
	private void addProessThread(){
		multiThread.add(new ProcessThread());
	}
	private void removeProessThread(){
		multiThread.remove(0).stopProcessThread=true;
	}
	private void stopMultiProcess(){
		if(multiThread!=null){
			while(multiThread.size()>0){
				multiThread.remove(0).stopProcessThread=true;
			}
		}
	}
	
	@Override
	public void run(){
		log.info(this.handlerName+" "+this.handlerId + " has start.");
		try{init();}catch(Throwable e){}
		while(!bStop || !queueMsg.isEmpty()){
			_processMsg();
		}
		try{stopNextHandlerList();}catch(Throwable e){}//当前处理流程结束后结束后续流程
		try{log();}catch(Throwable e){}
		try{stopMultiProcess();}catch(Throwable e){}
		try{stopped();}catch(Throwable e){}
		log.info(this.handlerName+" "+this.handlerId + " has stopped.");
		//清理
		try{ synchronized(BasicMsgHandler.class){ allHandlers.remove(this); if(allHandlers.isEmpty()) BasicMsgHandler.class.notifyAll(); } }catch(Throwable e){}
		handlerId = null;
		handlerName=null;
		nextHandlerList=null;
		parentHandlerList=null;
		entrCount=null;
		handlerCount=null;
		errorCount=null;
		queueMsg=null;
		multiThread=null;
		atomMultID=null;
	}
	
}